package com.patsnap.data.npd.dw.etl.job.ods.serialization;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.io.IOException;

@Slf4j
public class PreOdsInfoDeserializationSchema implements KafkaRecordDeserializationSchema<PreOdsInfo> {
    private static final long serialVersionUID = 1509391548173891955L;

    @Override
    public void deserialize(ConsumerRecord<byte[], byte[]> consumerRecord, Collector<PreOdsInfo> collector) throws IOException {
        String value = new String(consumerRecord.value());
        PreOdsInfo preOdsInfo = null;
        try {
            preOdsInfo = JSON.parseObject(value, PreOdsInfo.class);
            if (null != preOdsInfo) {
                collector.collect(preOdsInfo);
            }
        } catch (Exception e) {
            log.error("record format error:", e);
        }
    }

    @Override
    public TypeInformation<PreOdsInfo> getProducedType() {
        return TypeInformation.of(PreOdsInfo.class);
    }
}
